/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package aiaudio.processing.prediction.top;

import aiaudio.processing.base.mapreduce.CoordinateDoubleValueKey;
import aiaudio.lastfm.hbase.CannotCreateTableException;
import aiaudio.Application;
import aiaudio.processing.MainTableGroup;
import java.io.IOException;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.mapreduce.Job;

/**
 *
 * @author nastya
 */
public class MakeRecommendationAlgorithm{
    private final MainTableGroup mainTableGroup;

    public MakeRecommendationAlgorithm(MainTableGroup mainTableGroup) {
        this.mainTableGroup = mainTableGroup;
    }    
    
    private int maxPositionCount = 10;

    public void setMaxPositionCount(int maxPositionCount) {
        this.maxPositionCount = maxPositionCount;
    }
    

    
    public void start() throws IOException, CannotCreateTableException, InterruptedException, ClassNotFoundException {
        Job job = new Job();
        
        job.getConfiguration().setInt(MakeRecommendationReducer.MAX_POSITION_PARAM, maxPositionCount);
        
        Scan scan = new Scan();
        
        String recalcRatingsTable = Application.database().getByName(mainTableGroup.getRatingPredictionAlgTableGroup().getRecommendationCreationRatingList()).getDatabaseName();
        TableMapReduceUtil.initTableMapperJob(recalcRatingsTable, scan, CollectUserRatingsMapper.class,
                ImmutableBytesWritable.class, CoordinateDoubleValueKey.class, job);
        
        String resultTable = Application.database().getByName(mainTableGroup.getRatingPredictionAlgTableGroup().getPredictionsList()).getDatabaseName();
        TableMapReduceUtil.initTableReducerJob(resultTable, MakeRecommendationReducer.class, job);
        
        job.waitForCompletion(true);
    }

}
